package com.winit.schedule.job.tracker.event.persist;

import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.winit.schedule.core.conf.Configuration;
import com.winit.schedule.core.event.EventHandler;
import com.winit.schedule.core.repository.constants.LogType;
import com.winit.schedule.core.repository.constants.TaskItemStatus;
import com.winit.schedule.core.repository.entity.TaskItemEntity;
import com.winit.schedule.core.repository.entity.TaskTrackerEntity;
import com.winit.schedule.core.repository.query.TaskItemQuery;
import com.winit.schedule.core.utils.Clock;
import com.winit.schedule.job.tracker.JobTrackerContext;
import com.winit.schedule.job.tracker.conf.JobConfiguration;
import com.winit.schedule.job.tracker.repository.ScheduleDataManager;

public class TaskTrackerHandler implements EventHandler<DataPersistEvent> {

    private static final Logger             logger            = LoggerFactory.getLogger(TaskTrackerHandler.class);

    private static long                     defaultTimeout = 1000 * 60 * 3;

    private ScheduleDataManager             scheduleDataManager;

    private JobTrackerContext               jobTrackerContext;

    private List<Thread>                    dataHandlerThread = new ArrayList<Thread>();

    private BlockingQueue<DataPersistEvent> dataQueue         = new LinkedBlockingQueue<DataPersistEvent>();

    public TaskTrackerHandler(JobTrackerContext jobTrackerContext){
        this.jobTrackerContext = jobTrackerContext;
        this.scheduleDataManager = jobTrackerContext.getScheduleDataManager();
        this.defaultTimeout = jobTrackerContext.getConf().getLong(JobConfiguration.RECOVERY_TIMEOUT_MS,
                JobConfiguration.DEFAULT_RECOVERY_TIMEOUT_MS);

        for (int i = 1; i <= 2; i++) {
            Thread thread = new DataHandler(i);
            dataHandlerThread.add(thread);
            thread.start();
        }
    }

    @Override
    public void handle(DataPersistEvent event) {
        if (DataPersistEventType.REG_TASK_TRACKER.equals(event.getType()) && null != event.getTaskTracker()) {
            dataQueue.offer(event);
        }
    }

    public ScheduleDataManager getScheduleDataManager() {
        return scheduleDataManager;
    }

    public JobTrackerContext getJobTrackerContext() {
        return jobTrackerContext;
    }

    @SuppressWarnings("unchecked")
    public void handlerExceptionTask(String taskTrackerUUID, List<String> taskItemList) {
        if (null == taskItemList) {
            taskItemList = Collections.emptyList();
        }

        Configuration conf = jobTrackerContext.getConf();
        long timeout = conf.getLong(JobConfiguration.HEARTBEAT_TIMEOUT_MS,
            JobConfiguration.DEFAULT_HEARTBEAT_TIMEOUT_MS);
        TaskItemQuery query = new TaskItemQuery();
        query.setTimeout(Clock.getSystemTime() - (defaultTimeout > timeout ? defaultTimeout : timeout));
        query.setTaskItemStatusList(TaskItemStatus.RUN_STATUS);
        query.setLastFetchTaskTrackerUuid(taskTrackerUUID);
        List<TaskItemEntity> taskItems = scheduleDataManager.queryTaskItemsByTimeout(query);

        if (null != taskItems && taskItems.size() > 0) {
            TaskItemEntity item = null;
            for (int i = (taskItems.size() - 1) ; i >= 0; i--) {
                item = taskItems.get(i);
                if (taskItemList.contains(item.getTaskItemId())) {
                    taskItems.remove(item);
                }
            }
            if (taskItems.size() > 0) {
                jobTrackerContext.getDataDispatcher()
                    .getEventHandler()
                    .handle(new DataPersistEvent(DataPersistEventType.RECOVERY_TASKITEM,
                        null,
                        taskItems,
                        TaskItemStatus.STOP));
            }
        }
    }

    class DataHandler extends Thread {

        public DataHandler(int num){
            this.setName("Thread-TaskTracker-Persist-" + num);
        }

        @SuppressWarnings("static-access")
        @Override
        public void run() {
            while (true) {
                try {
                    DataPersistEvent event = dataQueue.take();
                    TaskTrackerEntity taskTracker = event.getTaskTracker();
                    if (null != taskTracker) {
                        String taskTrackerUUID = String.valueOf(taskTracker.getTaskTrackerUuid());
                        synchronized (taskTrackerUUID) {
                            TaskTrackerEntity oldTaskTracker = scheduleDataManager.getTaskTrackerByUuId(taskTrackerUUID);
                            if (null == oldTaskTracker) {
                                scheduleDataManager.saveOrUpdateTaskTrackerAndUpdateTaskInterfaceType(taskTracker);
                                scheduleDataManager.logger(taskTracker.getTaskTrackerUuid(),
                                    LogType.TASKTRACKER_CONNCET);
                            } else {
                                scheduleDataManager.updateHeartbeatTime(taskTracker);
                                taskTracker.setRegisterTime(oldTaskTracker.getRegisterTime());
                            }
                            
                            // 处理异常任务项
                            handlerExceptionTask(taskTracker.getTaskTrackerUuid(), event.getTaskItemIdList());
                        }
                    }

                    Thread.currentThread().sleep(50);
                } catch (Exception e) {
                    logger.error("TaskTrackerEntity data persist failure.", e);
                }
            }
        }
    }

}
